package hamster.csustef.acquisition;

import hamster.csustef.acquisition.constant.AcquisitionConstant;
import hamster.csustef.acquisition.flink.sink.HdfsSinkFactory;
import hamster.csustef.acquisition.flink.sink.KafkaProducerSinkFactory;
import hamster.csustef.acquisition.flink.source.DormKeySource;
import hamster.csustef.acquisition.flink.source.WeatherTaskSource;
import hamster.csustef.acquisition.flink.transfer.GetFeeMapFunction;
import hamster.csustef.acquisition.flink.transfer.GetWeatherMapFunction;
import hamster.csustef.acquisition.schema.MySimpleStringSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class AcquisitionApplication {
    public static void main(String[] args) throws Exception {
        int runningHour;
        try {
            runningHour = Integer.parseInt(args[0]);
            System.out.println("运行参数读取成功！");
        } catch (Exception ignore) {
            System.out.println("运行参数有误，使用默认参数！");
            runningHour = 1;
        }
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();


        // 电表爬虫
        // 1.1.1 读取寝室键数据，下发电费爬虫任务
        DataStreamSource<String> dromKeySource = env.addSource(new DormKeySource(runningHour));
        // 1.2.1 爬取电费数据，下发电费数据
        SingleOutputStreamOperator<String> getFee = dromKeySource.map(new GetFeeMapFunction());
        // 1.3.1 发送数据至Kafka
        getFee.addSink(KafkaProducerSinkFactory.getStringKafkaProducerSink(
                AcquisitionConstant.KAFKA_SERVERS,
                AcquisitionConstant.ELECTRIC_FEE_TOPIC));
        // 1.3.2 发送数据至HDFS
        getFee.sinkTo(HdfsSinkFactory.getHdfsSink(AcquisitionConstant.ELECTRIC_FEE_TEMP_FILE_PATH));


        // 天气爬虫
        // 2.1.1 下发天气爬虫任务
        DataStreamSource<String> weatherTaskSource = env.addSource(new WeatherTaskSource(runningHour));
        // 2.2.1 爬取天气数据，下发天气数据
        SingleOutputStreamOperator<String> getWeather = weatherTaskSource.map(new GetWeatherMapFunction());
        // 2.3.1 发送天气状态数据至Kafka
        getWeather.addSink(KafkaProducerSinkFactory.getStringKafkaProducerSink(
                AcquisitionConstant.KAFKA_SERVERS,
                AcquisitionConstant.CS_WEATHER_STATE_TOPIC));

        // 3.1.1 从Kafka中读取天气数据
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "master:9092");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                AcquisitionConstant.CS_WEATHER_TOPIC,
                new MySimpleStringSchema(),
                props
        );

        DataStreamSource<String> weatherDataSource = env.addSource(kafkaConsumer).setParallelism(1);
        // 3.1.2 将天气数据写入HDFS
        weatherDataSource.sinkTo(HdfsSinkFactory.getHdfsSink(AcquisitionConstant.CS_WEATHER_TEMP_FILE_PATH));

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        env.execute();

    }

}
